package com.ztesoft.web.backup.service.impl;


import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.ztesoft.core.threadpool.GeneralThreadPool;
import com.ztesoft.core.threadpool.ThreadPoolFactory;
import com.ztesoft.framework.log.ZTEsoftLogManager;
import com.ztesoft.framework.util.Utils;
import com.ztesoft.web.backup.BackUpTask;
import com.ztesoft.web.backup.BackUpThread;
import com.ztesoft.web.backup.service.BackUpService;
import com.ztesoft.web.baseconfig.db.dao.DmStrategyDao;
import com.ztesoft.web.baseconfig.db.po.DmStrategyPO;
import com.ztesoft.web.common.DMSConstant;
import com.ztesoft.web.common.customdb.dao.DmTaskStageCustomDao;
import com.ztesoft.web.taskgenerator.db.arg.DmTaskStageArg;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskStageDao;
import com.ztesoft.web.taskgenerator.db.po.DmTaskStagePO;

/**
 * 备份SERVICE
 * @author Chen.shitian
 * @date 2015年8月1日
 * @version 1.0
 */
@Service
public class BackUpServiceImpl implements BackUpService {
	
	/**
	 * 日志
	 */
    private final static ZTEsoftLogManager LOGGER = ZTEsoftLogManager
            .getLogger(BackUpServiceImpl.class);

	@Autowired
	private DmTaskStageDao dmTaskStageDao;//任务环节
	@Autowired
	private DmStrategyDao strategyDao;//策略
	private Queue<DmTaskStagePO> queue;//定义队列存放要备份的数据
	@Autowired
	private DmTaskStageCustomDao taskStageCustomDao;

	@Override
	public String backUpProcess(int taskId, int strategyId){
		String backUpState = "00A";
		LOGGER.info("|++===备份服务|任务ID="+taskId+",策略ID="+strategyId+"BEGIN..");
		if (Utils.isEmpty(taskId) || Utils.isEmpty(strategyId)) {
			throw new RuntimeException("传入的任务标志或策略标志为空");
		}
		LOGGER.debug("根据任务标识和策略标识查找任务环节表中待备份的数据,任务ID="+taskId+",策略ID="+strategyId);
		//根据任务标识和策略标识查找任务环节表中审核通过待备份的数据
		//Select * from dm_task_stage where dm_task_id=? And dm_strategy_id=? And stage=1 And state=’00T’
		DmTaskStageArg taskStageArg = new DmTaskStageArg();
		DmTaskStageArg.DmTaskStageCriteria c = taskStageArg.createCriteria();
		c.andDmTaskIdEqualTo(taskId);
		c.andDmStrategyIdEqualTo(strategyId);
		c.andStageEqualTo(DMSConstant.STAGE_BACKUP);
		c.andStateEqualTo(DMSConstant.Status.PENDING.toString());
		c.andIsCheckEqualTo(DMSConstant.IS_CHECK_PASS);
		List<DmTaskStagePO> taskStageList = dmTaskStageDao.selectByArg(taskStageArg);
		LOGGER.debug("获取策略信息,任务ID="+taskId+",策略ID="+strategyId);
		//获取策略信息
		DmStrategyPO strategyPO = strategyDao.selectByPrimaryKey(strategyId);
		//判断是否有带备份的数据
		GeneralThreadPool pool = null;
		LOGGER.info(Thread.currentThread()+"待执行备份的环节数{0}",taskStageList.size());
		if(taskStageList.size()>0) {
			queue = new ArrayBlockingQueue<DmTaskStagePO>(taskStageList.size());
			queue.addAll(taskStageList);
			List<Future<String>> futureList = new ArrayList<Future<String>>(taskStageList.size());
			int queueSize = queue.size();
			int coreSize = strategyPO.getConcurrentNumber() > queueSize ? queueSize : strategyPO.getConcurrentNumber();
            try {
            	//根据策略表（dm_strategy）的并发数（concurrent_number）创建备份线程池，调用备份处理流程，备份数据
	            pool = ThreadPoolFactory.createGeneralBoundedThreadPool("back up {taskId="+taskId+",strategyId="+strategyId+"}...", coreSize, queueSize);
    		} catch (Exception e) {
    			LOGGER.error("\r\n|获取数据源异常|任务ID="+taskId+",策略ID="+strategyId,e);
    			return DMSConstant.Status.FALSE.toString();
    		}
//            for (int i = 0; i <queueSize ; i++) {
            for (int i = 0; i <taskStageList.size() ; i++) {
            	if(DMSConstant.DATA_TYPE_TABLE == taskStageList.get(i).getDataType()){
            		BackUpThread thread = new BackUpThread(taskStageList.get(i),strategyPO.getBatchCommitCount());
            		Future<String> future = pool.submit(thread);
            		futureList.add(future);
            	}else{
            		BackUpTask task = new BackUpTask(taskStageList.get(i),strategyPO.getBatchCommitCount());
            		Future<String> future = pool.submit(task);
            		futureList.add(future);
            	}
             }
            int sucessInt = 0,failInt = 0,stopInt = 0, sysStopInt = 0, cancelInt = 0;//变量
    		for (Future<String> future : futureList) {
    			try {
    				if (DMSConstant.Status.SUCCESS.toString().equals(future.get())) {
						sucessInt = sucessInt + 1;
					} else if (DMSConstant.Status.FALSE.toString().equals(future.get())){
						failInt = failInt + 1;
					} else if (DMSConstant.Status.CANCEL.toString().equals(future.get())) {
						cancelInt = cancelInt + 1;
					} else if (DMSConstant.Status.PAUSE.toString().equals(future.get())) {
						stopInt = stopInt + 1;
					} else if (DMSConstant.Status.SYSPAUSE.toString().equals(future.get())) {
						sysStopInt = sysStopInt + 1;
					}
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			} catch (ExecutionException e) {
    				e.printStackTrace();
    			}
    		}
//    		if (pool.getCompletedTaskCount() == futureList.size()) {
			LOGGER.info(Thread.currentThread()+"backup 任务ID="+taskId+",策略ID="+strategyId+" end...");
			LOGGER.info("{0}|++===备份服务|任务END..sucessInt:"+sucessInt+",failInt:"+failInt+",stopInt:"+stopInt+"sysStopInt:"+sysStopInt+"cancelInt,:"+cancelInt,Thread.currentThread());
			if (cancelInt > 0){
				backUpState = "00C";//取消
			} else if (sysStopInt > 0) {
				backUpState = "02P";//系统暂停
			} else if (stopInt > 0) {
				backUpState = "00P";//暂停
			} else if (failInt > 0/*== futureList.size()*/) {
				backUpState = "00F";//失败
			} else {					
				backUpState = "00A";//成功
			}
			pool.shutDown();    	
		}
//		} 
		return backUpState;
	}

	
}
